#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>

namespace qqmq
{
// 定义数据文件和临时文件的后缀名
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"

    // 使用智能指针管理 Message 对象
    using MessagePtr = std::shared_ptr<qqmq::Message>;

    // MessageMapper 类负责消息的持久化存储和管理
    class MessageMapper
    {
    public:
        // 构造函数，初始化队列名称、数据文件和临时文件路径
        MessageMapper(std::string &basedir, const std::string &qname) : _qname(qname)
        {
            // 如果基础目录路径末尾没有 '/'，则添加 '/'
            if (basedir.back() != '/')
                basedir.push_back('/');
            // 拼接数据文件和临时文件的完整路径
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            // 检查基础目录是否存在，若不存在则创建
            if (FileHelper(basedir).exists() == false)
            {
                assert(FileHelper::createDirectory(basedir));
            }
            // 创建消息文件
            createMsgFile();
        }

        // 创建消息文件，如果文件已存在则直接返回 true
        bool createMsgFile()
        {
            if (FileHelper(_datafile).exists() == true)
            {
                return true;
            }
            // 创建文件
            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                // 记录日志，创建文件失败
                DLOG("创建队列数据文件 %s 失败！", _datafile.c_str());
                return false;
            }
            return true;
        }

        // 删除消息文件和临时文件
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        // 向数据文件插入消息
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }

        // 从数据文件中删除消息
        bool remove(MessagePtr &msg)
        {
            // 1. 将 msg 中的有效标志位修改为 '0'，表示消息无效
            msg->mutable_payload()->set_valid("0");
            // 2. 对 msg 进行序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                // 记录日志，新生成的数据与原数据长度不一致，无法修改
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
                return false;
            }
            // 3. 将序列化后的消息，写入到数据在文件中的指定位置（覆盖原有的数据）
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                // 记录日志，写入数据失败
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            // DLOG("确认消息后，删除持久化消息成功：%s", msg->payload().body().c_str());
            return true;
        }

        // 垃圾回收，清理无效消息
        std::list<MessagePtr> gc()
        {
            bool ret;
            std::list<MessagePtr> result;
            // 1. 加载文件中的有效数据
            ret = load(result);
            if (ret == false)
            {
                // 记录日志，加载有效数据失败
                DLOG("加载有效数据失败！\n");
                return result;
            }
            // DLOG("垃圾回收，得到有效消息数量：%d", result.size());
            // 2. 将有效数据，进行序列化存储到临时文件中
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                // DLOG("向临时文件写入数据: %s", msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    // 记录日志，向临时文件写入消息数据失败
                    DLOG("向临时文件写入消息数据失败！！");
                    return result;
                }
            }
            // DLOG("垃圾回收后，向临时文件写入数据完毕，临时文件大小: %ld", FileHelper(_tmpfile).size());
            // 3. 删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                // 记录日志，删除源文件失败
                DLOG("删除源文件失败！");
                return result;
            }
            // 4. 修改临时文件名，为源文件名称
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                // 记录日志，修改临时文件名称失败
                DLOG("修改临时文件名称失败！");
                return result;
            }
            // 5. 返回新的有效数据
            return result;
        }

    private:
        // 从文件中加载有效数据
        bool load(std::list<MessagePtr> &result)
        {
            // 1. 加载出文件中所有的有效数据；  存储格式 4字节长度|数据|4字节长度|数据.....
            FileHelper data_file_helper(_datafile);
            size_t offset = 0, msg_size;
            size_t fsize = data_file_helper.size();
            // 记录日志，准备开始加载持久化数据
            DLOG("准备开始加载持久化数据，当前文件大小: %ld", data_file_helper.size());
            bool ret;
            int i = 0;
            while (offset < fsize)
            {
                // 读取消息长度
                ret = data_file_helper.read((char *)&msg_size, offset, sizeof(size_t));
                // DLOG("第%d次读取数据,读取结果是%d",i, ret);
                i++;
                if (ret == false)
                {
                    // 记录日志，读取消息长度失败
                    DLOG("读取消息长度失败！");
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size, '\0');
                // 读取消息数据
                data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    // 记录日志，读取消息数据失败
                    DLOG("读取消息数据失败！");
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<Message>();
                // 解析消息数据
                msgp->mutable_payload()->ParseFromString(msg_body);
                // DLOG("加载到有效数据：%s", msgp->payload().body().c_str());
                // 如果是无效消息，则直接处理下一个
                if (msgp->payload().valid() == "0")
                {
                    // DLOG("加载到无效消息：%s", msgp->payload().body().c_str());
                    continue;
                }
                // 有效消息则保存起来
                result.push_back(msgp);
            }
            return true;
        }

        // 向指定文件插入消息
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增数据都是添加在文件末尾的
            // 1. 进行消息的序列化，获取到格式化后的消息
            std::string body = msg->payload().SerializeAsString();
            // 2. 获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();
            // 写入逻辑：1. 先写入4字节数据长度， 2， 再写入指定长度数据
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                // 记录日志，向队列数据文件写入数据长度失败
                DLOG("向队列数据文件写入数据长度失败！");
                return false;
            }
            // 3. 将数据写入文件的指定位置
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                // 记录日志，向队列数据文件写入数据失败
                DLOG("向队列数据文件写入数据失败！");
                return false;
            }
            // 4. 更新 msg 中的实际存储信息
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }

    private:
        std::string _qname;    // 队列名称
        std::string _datafile; // 数据文件路径
        std::string _tmpfile;  // 临时文件路径
    };

    // QueueMessage 类负责管理队列中的消息
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

        // 构造函数，初始化消息映射器、队列名称、有效消息数量和总消息数量
        QueueMessage(std::string &basedir, const std::string &qname) : _mapper(basedir, qname),
                                                                       _qname(qname), _valid_count(0), _total_count(0) {}

        // 恢复历史消息
        bool recovery()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            // 调用消息映射器的垃圾回收方法，获取有效消息
            _msgs = _mapper.gc();
            for (auto &msg : _msgs)
            {
                // 将消息添加到持久化消息哈希表中
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 更新有效消息数量和总消息数量
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        // 向队列中插入消息
        bool insert(const BasicProperties *bp, const std::string &body, DeliveryMode queue_is_durable)
        {
            // 1. 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                // 根据队列持久化模式设置消息的持久化模式
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                // 根据队列持久化模式设置消息的持久化模式
                DeliveryMode mode = (queue_is_durable == DeliveryMode::DURABLE) ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            // 2. 判断消息是否需要持久化
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 在持久化存储中表示数据有效
                // 3. 进行持久化存储
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    // 记录日志，持久化存储消息失败
                    DLOG("持久化存储消息：%s 失败了！", body.c_str());
                    return false;
                }
                // 持久化信息中的数据量 +1
                _valid_count += 1;
                _total_count += 1;
                // 将消息添加到持久化消息哈希表中
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 4. 内存的管理，将消息添加到待推送消息列表中
            _msgs.push_back(msg);
            return true;
        }

        // 获取队列的队首消息
        MessagePtr front()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }
            // 获取一条队首消息：从 _msgs 中取出数据
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将该消息对象，向待确认的哈希表中添加一份，等到收到消息确认后进行删除
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        // 每次删除消息后，判断是否需要垃圾回收
        bool remove(const std::string &msg_id)
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 从待确认队列中查找消息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                // 记录日志，没有找到要删除的消息
                DLOG("没有找到要删除的消息：%s!", msg_id.c_str());
                return true;
            }
            // 2. 根据消息的持久化模式，决定是否删除持久化信息
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3. 删除持久化信息
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                // 持久化文件中有效消息数量 -1
                _valid_count -= 1;
                // 内部判断是否需要垃圾回收，需要的话则回收一下
                gc();
            }
            // 4. 删除内存中的信息
            _waitack_msgs.erase(msg_id);
            // DLOG("确认消息后，删除消息的管理成功：%s", it->second->payload().body().c_str());
            return true;
        }

        // 获取待推送消息的数量
        size_t getable_count()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        // 获取总消息数量
        size_t total_count()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        // 获取持久化消息的数量
        size_t durable_count()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        // 获取待确认消息的数量
        size_t waitack_count()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        // 清空队列消息
        void clear()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            // 删除消息文件
            _mapper.removeMsgFile();
            // 清空待推送消息列表
            _msgs.clear();
            // 清空持久化消息哈希表
            _durable_msgs.clear();
            // 清空待确认消息哈希表
            _waitack_msgs.clear();
            // 重置有效消息数量和总消息数量
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        // 检查是否需要进行垃圾回收
        bool GCCheck()
        {
            // 持久化的消息总量大于 2000， 且其中有效比例低于 50% 则需要垃圾回收
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }

        // 进行垃圾回收
        void gc()
        {
            // 1. 进行垃圾回收，获取到垃圾回收后，有效的消息信息链表
            if (GCCheck() == false)
                return;
            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    // 记录日志，垃圾回收后，有一条持久化消息，在内存中没有进行管理
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    // 做法：重新添加到推送链表的末尾
                    _msgs.push_back(msg);
                    // 将消息添加到持久化消息哈希表中
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                // 2. 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 3. 更新当前的有效消息数量 & 总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;                                         // 互斥锁，用于线程安全
        std::string _qname;                                        // 队列名称
        size_t _valid_count;                                       // 持久化文件中有效消息的数量
        size_t _total_count;                                       // 持久化文件中总的消息数量
        MessageMapper _mapper;                                     // 消息映射器，负责消息的持久化操作
        std::list<MessagePtr> _msgs;                               // 待推送消息列表
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息哈希表，键为消息 ID，值为消息指针
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息哈希表，键为消息 ID，值为消息指针
    };

    // MessageManager 类负责管理多个队列的消息
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;

        // 构造函数，初始化基础目录
        MessageManager(const std::string &basedir) : _basedir(basedir) {}

        // 清空所有队列的消息
        void clear()
        {
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &qmsg : _queue_msgs)
            {
                // 调用每个队列消息对象的 clear 方法清空消息
                qmsg.second->clear();
            }
        }

        // 初始化指定队列的消息管理
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    // 若队列已存在，则直接返回
                    return;
                }
                // 创建队列消息对象
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                // 将队列消息对象插入到队列消息映射表中
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            // 恢复该队列的历史消息
            qmp->recovery();
        }

        // 销毁指定队列的消息管理
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 若队列不存在，则直接返回
                    return;
                }
                qmp = it->second;
                // 从队列消息映射表中移除该队列
                _queue_msgs.erase(it);
            }
            // 清空该队列的消息
            qmp->clear();
        }

        // 向指定队列插入消息
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, DeliveryMode mode)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，插入失败
                    DLOG("向队列%s新增消息失败：没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 insert 方法插入消息
            return qmp->insert(bp, body, mode);
        }

        // 获取指定队列的队首消息
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，获取失败
                    DLOG("获取队列%s队首消息失败：没有找到消息管理句柄!", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 front 方法获取队首消息
            return qmp->front();
        }

        // 确认指定队列中指定消息的处理结果
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，确认失败
                    DLOG("确认队列%s消息%s失败：没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 remove 方法删除该消息
            qmp->remove(msg_id);
            return;
        }

        // 获取指定队列待推送消息的数量
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，获取失败
                    DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 getable_count 方法获取待推送消息数量
            return qmp->getable_count();
        }

        // 获取指定队列总持久化消息的数量
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，获取失败
                    DLOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 total_count 方法获取总持久化消息数量
            return qmp->total_count();
        }

        // 获取指定队列有效持久化消息的数量
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，获取失败
                    DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 durable_count 方法获取有效持久化消息数量
            return qmp->durable_count();
        }

        // 获取指定队列待确认消息的数量
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                // 加锁保证线程安全
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 记录日志，未找到队列的消息管理句柄，获取失败
                    DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            // 调用队列消息对象的 waitack_count 方法获取待确认消息数量
            return qmp->waitack_count();
        }

    private:
        std::mutex _mutex;                                              // 互斥锁，用于线程安全
        std::string _basedir;                                           // 基础目录，用于存储队列消息文件
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; // 队列消息映射表，键为队列名称，值为队列消息对象指针
    };
}
#endif